package dowork;

import com.alibaba.fastjson.JSON;
import functions.JsontoBeanFunction;
import functions.MyversionV2KeyedProcessFunc;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import pojo.LogBean;
import pojo.ResultBean;
import source.KafkaSourceUtils;

import java.util.Properties;

public class RuleEngineV1 {

    public static void main(String[] args) throws Exception {
        DataStream<String> kafkaSource = KafkaSourceUtils.createKafkaSource();
        System.out.println("创建kfk流成功");
        KeyedStream<LogBean, String> logBeanStringKeyedStream = kafkaSource.map(new JsontoBeanFunction()).keyBy(new KeySelector<LogBean, String>() {
            @Override
            public String getKey(LogBean value) throws Exception {
                return value.getDeviceId();
            }
        });

        SingleOutputStreamOperator<ResultBean> result = logBeanStringKeyedStream.process(new MyversionV2KeyedProcessFunc());


        result.print();

        KafkaSourceUtils.env.execute("v2");



    }
}
